package com.link.serve.kafka.deprecated;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.ConsumerFactory;

public class DynamicKafkaConsumerFactory {

    /**
     * 构造一个 ConsumerFactory，根据传入参数动态生成消费者配置。
     * @param bootstrapServers Kafka 集群地址
     * @param groupId          消费组 ID
     * @param keyDeserializer  Key 的反序列化器，例如 org.apache.kafka.common.serialization.StringDeserializer.class
     * @param valueDeserializer Value 的反序列化器，例如 com.link.serve.kafka.coder.KafkaMessageDeserializer.class
     * @param extraConfigs     其他扩展配置，可以为空
     * @return ConsumerFactory 对象，泛型指定为 KafkaMessageProto.KafkaMessage（或你具体的消息类型）
     */
    public static ConsumerFactory<String, Object> createConsumerFactory(
            String bootstrapServers,
            String groupId,
            Class<?> keyDeserializer,
            Class<?> valueDeserializer,
            Map<String, Object> extraConfigs) {

        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);

        if (extraConfigs != null && !extraConfigs.isEmpty()) {
            props.putAll(extraConfigs);
        }

        return new DefaultKafkaConsumerFactory<>(props);
    }
}
